package felix.things.router.mqtt.server.broker.webapi;

import cn.hutool.core.codec.Base64;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import felix.things.router.mqtt.server.broker.cluster.RedisCluster;
import felix.things.router.mqtt.server.broker.config.BrokerProperties;
import felix.things.router.mqtt.server.broker.internal.InternalMessage;
import felix.things.router.mqtt.server.broker.internal.InternalSendServer;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.client.RestTemplate;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.function.Consumer;

import static felix.things.router.mqtt.server.store.utils.StoreUtil.getProcessId;

/**
 * @author Felix
 * @date 2020/6/14
 */
@RestController
@RequestMapping("/open/api/mqtt")
@Slf4j
public class WebApiController {
    private static final String CACHE_SESSION_PRE = "mqtt-server:session:";
    private static final String CACHE_CLIENT_PRE = "mqtt-server:client:";
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    @Autowired
    private InternalSendServer internalSendServer;
    @Autowired
    private BrokerProperties brokerProperties;
    @Autowired
    private RedisCluster redisCluster;

    private RestTemplate restTemplate = new RestTemplate();

    /**
     * 向设备发送数据,发送格式见 test_send 方法实例代码
     *
     * @param data
     * @return
     */
    @PostMapping("/send")
    public JSONObject send(@RequestBody JSONObject data) {
        JSONObject result = new JSONObject();
        try {
            String processId = getProcessId("0");
            InternalMessage message = new InternalMessage();
            message.setBrokerId(brokerProperties.getId());
            message.setProcessId(processId);
            message.setClientId(UUID.randomUUID().toString());
            message.setTopic(data.getString("topic"));
            message.setRetain(data.getBoolean("retain"));
            message.setDup(data.getBoolean("dup"));
            message.setMqttQoS(data.getIntValue("qos"));
            message.setMessageBase64(Base64.encode(data.getString("message").getBytes()));
            log.debug("send:::" + JSON.toJSONString(message));
            //如果开启集群功能
            if (brokerProperties.isClusterEnabled()) {
                redisCluster.sendMessage(message);
            } else {
                internalSendServer.sendPublishMessage(message.getClientId(), message.getTopic(), MqttQoS.valueOf(message.getMqttQoS()),
                        Base64.decode(message.getMessageBase64()), message.isRetain(), message.isDup());
            }
            result.put("code", 0);
            result.put("msg", "success");
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            result.put("code", -1);
            result.put("msg", e.getMessage());
        }
        return result;
    }

    @GetMapping("/send")
    public Object test_send() {
        JSONObject result = new JSONObject();
        try {
            JSONObject message = new JSONObject();
            message.put("topic", "testtopic/1");
            message.put("retain", true);
            message.put("dup", true);
            message.put("qos", 1);
            message.put("message", "wizzer");
            ResponseEntity<String> resp = restTemplate.postForEntity("http://127.0.0.1:8080/open/api/mqtt/send", message, String.class);
            if (resp.getStatusCode().is2xxSuccessful()) {
                result.put("code", 0);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            result.put("code", -1);
        }
        return result;
    }

    /**
     * 获取在线设备数量、客户端名称、订阅主题
     * example: {"code":0,"msg":"","data":{"total":0,"list":[{"clientId":"pc-web","topics":["/topic_back"]}]}}
     */
    @GetMapping("/info")
    public Object info() {
        JSONObject result = new JSONObject();
        try {
            JSONObject data = new JSONObject();
            List<String> list = new ArrayList<>();
            this.scan(CACHE_SESSION_PRE + "*", item -> {
                //符合条件的key
                String key = new String(item, StandardCharsets.UTF_8);
                list.add(key);
            });
            List<JSONObject> dataList = new ArrayList<>();
            JSONObject item = new JSONObject();
            for (String k : list) {
                item.put("clientId", k.substring(k.lastIndexOf(":") + 1));
                item.put("topics", redisTemplate.opsForSet().members(CACHE_CLIENT_PRE + k.substring(k.lastIndexOf(":") + 1)));
                dataList.add(item);
            }
            data.put("total", list.size());
            data.put("list", dataList);
            result.put("code", 0);
            result.put("msg", "");
            result.put("data", data);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            result.put("code", -1);
            result.put("msg", e.getMessage());
        }
        return result;
    }

    /**
     * scan 实现
     *
     * @param pattern  表达式
     * @param consumer 对迭代到的key进行操作
     */
    public void scan(String pattern, Consumer<byte[]> consumer) {
        this.redisTemplate.execute((RedisConnection connection) -> {
            try (Cursor<byte[]> cursor = connection.scan(ScanOptions.scanOptions().count(Long.MAX_VALUE).match(pattern).build())) {
                cursor.forEachRemaining(consumer);
                return null;
            } catch (IOException e) {
                e.printStackTrace();
                throw new RuntimeException(e);
            }
        });
    }
}
